package com.icbc.dccsh.bolt;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisLookupMapper;

import com.google.common.collect.Lists;
import com.icbc.dccsh.util.Watcher;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.ITuple;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class GetHistoryMapper implements RedisLookupMapper {
	private RedisDataTypeDescription description;
	private Watcher _watcher = null;
	
	public GetHistoryMapper() {
        description = new RedisDataTypeDescription(
                RedisDataTypeDescription.RedisDataType.STRING, null);
        this._watcher = Watcher.getWatcher("GetHistoryMapper");
    }

	private static flexjson.JSONDeserializer d = new flexjson.JSONDeserializer();
	public List<Values> toTuple(ITuple input, Object value) {
		this._watcher.count();
		HashMap<String, Object> input_hashmap = (HashMap<String, Object>)input.getValueByField("input");
		HashMap<String, Object> mapping_hashmap = (HashMap<String, Object>)input.getValueByField("mapping");
		if (StringUtils.isEmpty((String)value)){
			System.out.println(String.format("mapping for key [%s] not found...", getKey(input)));
		}
		HashMap<String, Object> history_hashmap = (HashMap<String, Object>)d.deserialize((String)value);
		
		List<Values> values = Lists.newArrayList();
		values.add(new Values(input_hashmap, mapping_hashmap, history_hashmap));
    	return values;
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("input", "mapping", "history"));
	}

	public RedisDataTypeDescription getDataTypeDescription() {
		return description;
	}
	
	public String getKeyFromTuple(ITuple tuple) {
		return getKey(tuple);
//		HashMap<String, Object> input_hash = (HashMap<String, Object>) tuple.getValueByField("input");
//		HashMap endpoint = (HashMap) input_hash.get("endpoint");
//		String devicename = (String) endpoint.get("devicename");
//		String ip = (String) endpoint.get("ip");
//		String portname = (String) input_hash.get("object");
//
//		String key = String.format("msg:p:d:%s:ip:%s:o:%s", devicename, ip, portname);
//		return key;
	}

	private static String getKey(ITuple tuple){
		HashMap<String, Object> input_hash = (HashMap<String, Object>) tuple.getValueByField("input");
//		String key = (String)input_hash.get("mapping");
		String objectName = "unset";
		HashMap<String, HashMap<String, String>> objectNameHash = (HashMap<String, HashMap<String, String>>)input_hash.get("metrics");
		HashMap<String, String> objectNameHash2 = objectNameHash.get("Name");
		for (Map.Entry<String, String> entry : objectNameHash2.entrySet()){
			objectName = entry.getValue();
			break;
		}
		
		HashMap endpoint = (HashMap) input_hash.get("endpoint");
		String devicename = (String) endpoint.get("devName");
		String ip = (String) endpoint.get("devIP");
//		String portname = (String) input_hash.get("object");

		String key = String.format("msg:p:d:%s:ip:%s:o:%s", devicename, ip, objectName);
		
		return key;
	}
	
	@Override
	public String getValueFromTuple(ITuple tuple) {
		return null;
	}

//	public void execute(Tuple tuple, BasicOutputCollector collector) {
//		this._watcher.count();
//		HashMap<String, Object> input_hashmap = (HashMap<String, Object>)tuple.getValueByField("input");
//		HashMap<String, Object> mapping_hash = (HashMap<String, Object>)tuple.getValueByField("mapping");
//		boolean needHistory
//		
//		collector.emit(new Values(map));
//	}
	// public void declareOutputFields(OutputFieldsDeclarer declarer) {
	// declarer.declare(new Fields("hash"));
	// }
}
